package org.apache.solr.ocean.replication;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpMethod;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.RequestEntity;
import org.apache.commons.lang.ObjectUtils;
import org.apache.lucene.store.RAMDirectory;
import org.apache.solr.ocean.Documents;
import org.apache.solr.ocean.Loader;
import org.apache.solr.ocean.core.Deletes;
import org.apache.solr.ocean.core.Batch.MasterBatch;
import org.apache.solr.ocean.core.Batch.SerializableBatchDeletes;
import org.apache.solr.ocean.core.Batch.SerializableBatchDocuments;
import org.apache.solr.ocean.core.Batch.SerializableRamDirectory;
import org.apache.solr.ocean.core.Batch.SlaveBatch;
import org.apache.solr.ocean.http.HttpParameters;
import org.apache.solr.ocean.http.HttpTimeout;
import org.apache.solr.ocean.http.MulticastHttp;
import org.apache.solr.ocean.http.OceanXML;
import org.apache.solr.ocean.http.MulticastHttp.AllExceptions;
import org.apache.solr.ocean.http.MulticastHttp.ExceptionEnum;
import org.apache.solr.ocean.http.MulticastHttp.HttpRequestException;
import org.apache.solr.ocean.replication.ByteBufferPool.ByteBuffers;
import org.apache.solr.ocean.replication.LogFile2.Record;
import org.apache.solr.ocean.replication.LogFileManager.RecordIterator;
import org.apache.solr.ocean.replication.RawLogFile.ByteStreamRecord;
import org.apache.solr.ocean.replication.RawLogFile.RecordHeader;
import org.apache.solr.ocean.replication.RawLogFile.StreamRecord;
import org.apache.solr.ocean.util.ByteArrayOutputStream;
import org.apache.solr.ocean.util.LongSequence;
import org.apache.solr.ocean.util.RestartableTimer;
import org.apache.solr.ocean.util.Timeout;
import org.apache.solr.ocean.util.Util;
import org.apache.solr.ocean.util.RestartableTimer.TimeoutRunnable;
import org.apache.solr.schema.IndexSchema;

// TODO: reset sequences on master change
// TODO: move master name stuff out, create interface
// TODO: use static httpclient
public class Replication {
  public static Logger log = Logger.getLogger(Replication.class.getName());
  public static long LEASE_DURATION = 1000 * 5;
  private String id;
  private URL url;
  private LogFileManager logFileManager;
  private File replicationDirectoryFile;
  private LongSequence sequence;
  private Loader loader;
  private ReentrantLock writeLock = new ReentrantLock();
  private ByteBufferPool byteBufferPool = new ByteBufferPool(50 * 1024, 5, 5);
  //private ByteBufferPool deletesBytePool = new ByteBufferPool(5 * 1024, 5, 10);
  private Set<URL> nodeUrls = new HashSet<URL>();
  private String operatingRole;
  private String masterId;
  private ExecutorService threadPool;
  private Leases leases;
  private SlaveClient client;
  private Cell cell;
  private int serverNumber;

  public Replication(String id, int serverNumber, URL url, File replicationDirectoryFile, Loader loader, Cell cell) throws Exception,
      IOException {
    this.id = id;
    this.serverNumber = serverNumber;
    this.url = url;
    this.replicationDirectoryFile = replicationDirectoryFile;
    this.loader = loader;
    this.cell = cell;
    cell.registerListener(new CellHandler());
    logFileManager = new LogFileManager(replicationDirectoryFile);
    Long highestSnapshotId = logFileManager.getHighestId();
    Long sequenceId = Util.getNextServerSequence(highestSnapshotId, serverNumber);
    leases = new Leases(this);
    sequence = new LongSequence(sequenceId, 100);
    threadPool = Executors.newFixedThreadPool(6);
    client = new SlaveClient();
  }
  
  public NodeInfo getNodeInfo() {
    return new NodeInfo(id, url, masterId, operatingRole, getHighestId());
  }
  
  public LogFileManager getLogFileManager() {
    return logFileManager;
  }
  
  public String getNodeId() {
    return id;
  }

  public void resetSequence() {
    Long highestSnapshotId = getHighestId();
    Long sequenceId = Util.getNextServerSequence(highestSnapshotId, serverNumber);
    sequence.set(sequenceId);
  }

  public Long getPreviousId(Long id) {
    return logFileManager.getPreviousId(id);
  }

  public boolean contains(Long id) {
    return logFileManager.contains(id);
  }

  public Long getLowestId() {
    return logFileManager.getLowestId();
  }

  public Long getHighestId() {
    return logFileManager.getHighestId();
  }
  
  /**
  public Record getRecord(RecordHeader recordHeader) throws Exception {
    return logFileManager.getRecord(recordHeader);
  }

  public List<RecordHeader> getRecordHeaders(Long snapshotId, int num) throws Exception {
    return logFileManager.getRecordHeaders(snapshotId, num);
  }
  **/
  public static HttpTimeout getDefaultHttpTimeout() {
    return new HttpTimeout(1000, 1000);
  }

  public Leases getLeases() {
    return leases;
  }

  /**
   * Runs until all latest transactions downloaded
   */
  public class SlaveClient implements TimeoutRunnable {
    // private URL url;
    private NodeInfo nodeInfo;
    private RestartableTimer timer;

    public void setMaster(NodeInfo nodeInfo) throws InterruptedException {
      this.nodeInfo = nodeInfo;
      if (nodeInfo == null) {
        timer.cancel(false);
      } else {
        timer.runNow();
      }
    }

    public SlaveClient() throws Exception {
      timer = new RestartableTimer("replication", this, 5 * 1000, Long.MAX_VALUE);
      setMaster(null);
    }

    public void run(Timeout timeout) {
      if (url == null)
        return;
      try {
        while (true) {
          boolean renewed = renewLease();
          if (renewed) {
            break;
          } else {
            long startid = getHighestId() + 1;
            downloadTransactions(startid, 100);
          }
        }
      } catch (Exception exception) {
        log.info(exception.getMessage());
        cell.notifyMasterFailure(nodeInfo.getId());
        // runAsOperatingRole();
        // TODO: if master fails what to do?
      }
    }

    private void downloadTransactions(Long startid, int num) throws Exception {
      HttpClient httpClient = new HttpClient();
      HttpMethod httpMethod = new GetMethod(url + "?startid=" + startid + "&num=" + num);
      httpClient.executeMethod(httpMethod);
      int statusCode = httpMethod.getStatusCode();
      if (statusCode == 200) {
        DataInputStream dataInput = new DataInputStream(httpMethod.getResponseBodyAsStream());
        int size = dataInput.readInt();
        for (int x = 0; x < size; x++) {
          ByteBuffers documentsByteBuffers = null;
          ByteBuffers deletesByteBuffers = null;
          try {
            long id = dataInput.readLong();
            String category = dataInput.readUTF();
            long schemaVersion = dataInput.readInt();
            int documentsLength = dataInput.readInt();
            documentsByteBuffers = new ByteBuffers(dataInput, 1024 * 32, documentsLength, byteBufferPool);
            if (documentsByteBuffers.getLength() != documentsLength) {
              throw new Exception("documentsRead not equal to documentsLength");
            }
            int deletesLength = dataInput.readInt();
            deletesByteBuffers = new ByteBuffers(dataInput, 1024 * 4, deletesLength, byteBufferPool);
            if (deletesByteBuffers.getLength() != deletesLength) {
              throw new Exception("deletesRead not equal to deletesLength");
            }
            ByteStreamRecord streamRecord = new ByteStreamRecord(documentsByteBuffers, deletesByteBuffers);
            writeSlaveBatch(id, category, schemaVersion, streamRecord);
          } finally {
            if (documentsByteBuffers != null)
              documentsByteBuffers.finished();
            if (deletesByteBuffers != null)
              deletesByteBuffers.finished();
          }
        }
      } else {
        throw new Exception("server error");
      }
    }

    private boolean renewLease() throws Exception {
      HttpParameters parameters = new HttpParameters();
      parameters.add("action", "renewlease");
      NodeInfo node = getNodeInfo();
      node.transferTo(parameters);
      HttpTimeout httpTimeout = getDefaultHttpTimeout();
      OceanXML response = OceanXML.request(url, parameters, httpTimeout);
      return response.getHeaderBoolean("renewed");
    }
  }

  public boolean renewLease(NodeInfo node, long duration) throws Exception {
    if (!node.getHighestLogId().equals(logFileManager.getHighestId())) {
      return false;
    } else {
      leases.renewLease(node, duration);
      return true;
    }
  }

  public static class Leases {
    private Map<String,Lease> leaseMap = new HashMap<String,Lease>();
    private ReentrantLock lock = new ReentrantLock();
    private Replication replication;

    public Leases(Replication replication) {
      this.replication = replication;
    }

    private void cancelLease(URL url) {
      lock.lock();
      try {
        for (Lease lease : leaseMap.values()) {
          NodeInfo node = lease.getNodeInfo();
          if (node.getUrl().equals(url)) {
            cancelLease(node.getId());
          }
        }
      } finally {
        lock.unlock();
      }
    }

    private void cancelLease(String id) {
      leaseMap.remove(id);
      replication.cell.notifySlaveNotInSync(replication.getNodeId(), id);
    }

    private void renewLease(NodeInfo node, long duration) {
      lock.lock();
      try {
        String nodeid = node.getId();
        Lease lease = leaseMap.get(nodeid);
        if (lease == null) {
          lease = new Lease(node);
          leaseMap.put(nodeid, lease);
        }
        lease.renew(duration);
      } finally {
        lock.unlock();
      }
    }

    public List<NodeInfo> getNodeInfos() {
      lock.lock();
      try {
        List<NodeInfo> nodes = new ArrayList<NodeInfo>();
        long currentTime = System.currentTimeMillis();
        for (Map.Entry<String,Lease> entry : leaseMap.entrySet()) {
          Lease lease = entry.getValue();
          Long currentId = replication.getHighestId();
          if (lease.isExpired(currentTime) || !lease.isIdCurrent(currentId)) {
            cancelLease(entry.getKey());
          } else {
            nodes.add(lease.getNodeInfo());
          }
        }
        return nodes;
      } finally {
        lock.unlock();
      }
    }

    public static class Lease {
      private long expiration;
      private NodeInfo node;

      public Lease(NodeInfo node) {
        this.node = node;
      }

      public NodeInfo getNodeInfo() {
        return node;
      }

      public boolean isIdCurrent(Long id) {
        return ObjectUtils.equals(node.getHighestLogId(), id);
      }

      public boolean isExpired(long currentTime) {
        return currentTime > expiration;
      }

      public void renew(long duration) {
        expiration = duration + System.currentTimeMillis();
      }
    }
  }

  public ByteBufferPool getByteBufferPool() {
    return byteBufferPool;
  }

  public void writeSlaveBatch(Long id, String category, long schemaVersion, StreamRecord data) throws IOException {
    writeLock.lock();
    try {
      logFileManager.writeRecord(id, category, schemaVersion, data);
      resetSequence();
    } finally {
      writeLock.unlock();
    }
  }

  public class CellHandler implements Cell.Listener {
    public void newMaster(NodeInfo newMaster) {
      try {
        if (!newMaster.getId().equals(getNodeId())) {
          client.setMaster(newMaster);
        } else {
          client.setMaster(null);
        }
      } catch (InterruptedException ex) {
        log.log(Level.SEVERE, "", ex);
      }
    }
  }

  /**
   * public void setNodeUrls(Set<URL> newNodeUrls) { Set<URL> added = new
   * HashSet<URL>(newNodeUrls); added.removeAll(nodeUrls); Set<URL> removed =
   * new HashSet<URL>(nodeUrls); removed.removeAll(newNodeUrls); nodeUrls =
   * newNodeUrls; runAsOperatingRole(); }
   * 
   * public void runAsOperatingRole() { threadPool.submit(new
   * RunAsOperatingRole()); }
   * 
   * private class RunAsOperatingRole implements Runnable { public void run() {
   * try { Node master = findVotedMaster(new Timeout(1000 * 60 * 5)); if
   * (master.getId().equals(id)) { operatingRole = "master";
   * client.setUrl(null); notifyName(); } else { URL url = master.getUrl();
   * client.setUrl(url); operatingRole = "slave"; } } catch (Throwable
   * throwable) { log.log(Level.SEVERE, "", throwable); } } }
   * 
   * public List<Node> getNodes() { try { MulticastHttp multicastHttp = new
   * MulticastHttp(threadPool, ExceptionEnum.NONE); HttpTimeout
   * defaultHttpTimeout = getDefaultHttpTimeout(); HttpParameters parameters =
   * new HttpParameters("action", "getnode"); List<OceanXML> responses =
   * multicastHttp.execute(nodeUrls, parameters, defaultHttpTimeout); List<Node>
   * nodes = new ArrayList<Node>(); for (OceanXML response : responses) { Node
   * node = null; nodes.add(node); } return nodes; } catch (Throwable throwable) {
   * throw new RuntimeException(throwable); } }
   * 
   * private Node findVotedMaster(Timeout timeout) throws Throwable,
   * InterruptedException, TimeoutException { while (timeout.isOK()) { Node node =
   * findVotedMaster(); if (node != null) return node; } throw new
   * TimeoutException(timeout); }
   */
  public String getOperatingRole() {
    return operatingRole;
  }

  /**
   * private Node findVotedMaster() throws InterruptedException, Throwable {
   * MulticastHttp multicastHttp = new MulticastHttp(threadPool,
   * ExceptionEnum.NONE); HttpTimeout defaultHttpTimeout =
   * getDefaultHttpTimeout(); HttpParameters parameters = new
   * HttpParameters("action", "getchosenmaster"); List<OceanXML> responses =
   * multicastHttp.execute(nodeUrls, parameters, defaultHttpTimeout); if
   * (isMajority(responses.size())) { Map<String, Node> chosenMasterMap = new
   * HashMap<String, Node>(); for (OceanXML response : responses) { Node node =
   * new Node(response.getFirstInBody());
   * chosenMasterMap.put(node.getChosenMasterId(), node); } if
   * (chosenMasterMap.size() == 1) { return
   * chosenMasterMap.values().iterator().next(); } } return null; }
   * 
   * private boolean isMajority(int size) { if ((size / 2) > nodeUrls.size()) {
   * return true; } else return false; } /** public Node getNode() { String role =
   * getOperatingRole(); Long highestLogID = sequence.get(); String
   * chosenMasterID = null; if (chosenMaster != null) { chosenMasterID =
   * chosenMaster.getId(); } Node node = new Node(id, url, chosenMasterID, role,
   * highestLogID); return node; }
   */
  /**
   * public boolean isChosenMaster() { Node node = getChosenMaster(); if (node ==
   * null) return false; return StringUtils.equals(node.getId(), id); }
   * 
   * public Node getChosenMaster() { if (chosenMaster == null) { chosenMaster =
   * chooseMaster(getNodes()); } return chosenMaster; }
   * 
   * private static Node chooseMaster(List<Node> nodes) { int masterCount =
   * getMasterCount(nodes); if (masterCount > 1) return null; else if
   * (masterCount == 1) { return getMaster(nodes); } else if (masterCount == 0) {
   * try { ReflectColumns reflectColumns = new ReflectColumns();
   * reflectColumns.add("highestCellTransactionIDDatabase",
   * SortReflect.method("getHighestLogID"), SorterConstants.DESCENDING_ORDER);
   * reflectColumns.add("id", SortReflect.method("getId"),
   * SorterConstants.ASCENDING_ORDER); Sorter sorter =
   * Sorter.createReflect(reflectColumns, nodes,
   * SorterConstants.NULLS_ARE_LEAST); List<Node> sortedNodes =
   * sorter.sortReflect(); return sortedNodes.get(0); } catch (MethodException
   * methodException) { throw new RuntimeException(methodException.getCause()); } }
   * throw new RuntimeException("should not have reached"); }
   * 
   * private static Node getMaster(Collection<Node> nodes) { for (Node node :
   * nodes) { String role = node.getOperatingRole(); if
   * (StringUtils.equals(role, "master")) return node; } return null; }
   * 
   * private static int getMasterCount(Collection<Node> nodes) { int count = 0;
   * for (Node node : nodes) { String role = node.getOperatingRole(); if
   * (StringUtils.equals(role, "master")) count++; } return count; }
   */

  public static class ReplicationException extends Exception {
    public ReplicationException(String message, Throwable throwable) {
      super(message, throwable);
    }
  }

  public Long getNextId() {
    return sequence.getAndIncrement();
  }
  
  public SlaveBatchIterator getSlaveBatchIterator(String category, Long snapshotId) throws Exception {
    return new SlaveBatchIterator(category, snapshotId);
  }
  
  public class SlaveBatchIterator {
    RecordIterator recordIterator;
    
    public SlaveBatchIterator(Long snapshotId) throws IOException {
      recordIterator = logFileManager.getRecordIterator(snapshotId);
    }
    
    public SlaveBatchIterator(String category, Long snapshotId) throws Exception {
      recordIterator = logFileManager.getRecordIterator(category, snapshotId);
    }
    
    public boolean hasNext() throws IOException {
      return recordIterator.hasNext();
    }
    
    public SlaveBatch next(boolean loadDocuments, boolean loadOther, byte[] buffer) throws Exception {
      Record record = recordIterator.next();
      Documents documents = null;
      RAMDirectory ramDirectory = null;
      Deletes deletes = null;
      if (loadDocuments) {
        InputStream input = record.getStreamRecord().getDocuments().getInputStream(buffer);
        ObjectInputStream objectInputStream = new ObjectInputStream(input);
        Object deserialized = objectInputStream.readObject();
        if (deserialized instanceof SerializableRamDirectory) {
          SerializableRamDirectory serializableRamDirectory = (SerializableRamDirectory) deserialized;
          ramDirectory = serializableRamDirectory.getRamDirectory();
        } else if (deserialized instanceof SerializableBatchDocuments) {
          SerializableBatchDocuments serializableBatchDocuments = (SerializableBatchDocuments) deserialized;
          documents = serializableBatchDocuments.getDocuments();
        }
      }
      if (loadOther) {
        InputStream inputStream = record.getStreamRecord().getOther().getInputStream(buffer);
        ObjectInputStream deletesObjectInputStream = new ObjectInputStream(inputStream);
        SerializableBatchDeletes serializableBatchDeletes = (SerializableBatchDeletes) deletesObjectInputStream.readObject();
        deletes = serializableBatchDeletes.getDeletes();
      }
      String category = record.getCategory();
      float schemaVersion = record.getSchemaVersion();
      IndexSchema indexSchema = loader.getIndexSchema(category, schemaVersion, timeout);
      if (ramDirectory != null) {
        return new SlaveBatch(category, record.getId(), indexSchema, ramDirectory, deletes);
      } else {
        return new SlaveBatch(record.getCategory(), record.getId(), indexSchema, documents, deletes);
      }
    }
    
    public void close() throws IOException {
      recordIterator.close();
    }
  }
  
  /**
  public SlaveBatch getBatchGreaterOrEqualTo(String category, Long id) throws Exception {
    EnumSet<Load> load = EnumSet.of(Load.DOCUMENTS, Load.DELETES);
    return getBatchGreaterOrEqualTo(load, category, id);
  }

  public List<SlaveBatch> getBatchesGreaterOrEqualTo(EnumSet<Load> load, String category, Long id) throws Exception {
    List<SlaveBatch> slaveBatches = new ArrayList<SlaveBatch>();
    Long currentID = id;
    for (;;) {
      SlaveBatch slaveBatch = getBatchGreaterOrEqualTo(load, category, id);
      if (slaveBatch != null) {
        slaveBatches.add(slaveBatch);
        currentID = slaveBatch.getId() + 1;
      } else {
        break;
      }
    }
    return slaveBatches;
  }
  **/
  /**
  public SlaveBatch getBatchGreaterOrEqualTo(EnumSet<Load> load, String category, Long id) throws Exception {
    Record record = null;
    try {
      record = getRecordGreaterOrEqualTo(load, category, id);
      Data data = record.getData();
      Documents documents = null;
      RAMDirectory ramDirectory = null;
      Deletes deletes = null;
      if (load.contains(Load.DOCUMENTS)) {
        InputStream inputStream = data.getDocumentsByteBuffers().getInputStream();
        ObjectInputStream objectInputStream = new ObjectInputStream(inputStream);
        Object unserialized = objectInputStream.readObject();
        if (unserialized instanceof SerializableRamDirectory) {
          SerializableRamDirectory serializableRamDirectory = (SerializableRamDirectory) unserialized;
          ramDirectory = serializableRamDirectory.getRamDirectory();
        } else if (unserialized instanceof SerializableBatchDocuments) {
          SerializableBatchDocuments serializableBatchDocuments = (SerializableBatchDocuments) unserialized;
          documents = serializableBatchDocuments.getDocuments();
        }
      }
      if (load.contains(Load.DELETES)) {
        InputStream inputStream = data.getDeletesByteBuffers().getInputStream();
        ObjectInputStream deletesObjectInputStream = new ObjectInputStream(inputStream);
        SerializableBatchDeletes serializableBatchDeletes = (SerializableBatchDeletes) deletesObjectInputStream.readObject();
        deletes = serializableBatchDeletes.getDeletes();
      }
      long schemaVersion = record.getSchemaVersion();
      IndexSchema indexSchema = loader.getIndexSchema(category, schemaVersion, new Timeout(1000));
      if (ramDirectory != null) {
        return new SlaveBatch(category, id, indexSchema, ramDirectory, deletes);
      } else {
        return new SlaveBatch(category, id, indexSchema, documents, deletes);
      }
    } catch (ClassNotFoundException classNotFoundException) {
      IOException ioException = new IOException(classNotFoundException.getMessage());
      ioException.initCause(classNotFoundException);
      throw ioException;
    } finally {
      if (record != null) {
        Data data = record.getData();
        if (data != null)
          data.finished();
      }
    }
  }

  public Record getRecordGreaterOrEqualTo(Long id) throws IOException {
    EnumSet<Load> load = EnumSet.of(Load.DOCUMENTS, Load.DELETES);
    return logFileManager.getRecordGreaterOrEqualTo(load, id, documentsBytePool, deletesBytePool);
  }

  public Record getRecordGreaterOrEqualTo(EnumSet<Load> load, String category, Long id) throws IOException {
    return logFileManager.getRecordGreaterOrEqualTo(load, category, id, documentsBytePool, deletesBytePool);
  }
  **/
  public void writeMasterBatch(final Long id, final Long previousId, final MasterBatch masterBatch) throws Exception {
    final String category = masterBatch.getCategoryName();
    Serializable serializableDocuments = masterBatch.getSerializableBatchDocuments(id);
    Serializable serializableDeletes = masterBatch.getSerializableBatchDeletes(id);
    final ByteArrayOutputStream documentsByteOutput = new ByteArrayOutputStream(getByteBufferPool());
    final ByteArrayOutputStream deletesByteOutput = new ByteArrayOutputStream(getByteBufferPool());
    try {
      ObjectOutputStream documentsObjectOutputStream = new ObjectOutputStream(documentsByteOutput);
      documentsObjectOutputStream.writeObject(serializableDocuments);
      documentsObjectOutputStream.close();
      ObjectOutputStream deletesObjectOutputStream = new ObjectOutputStream(deletesByteOutput);
      deletesObjectOutputStream.writeObject(serializableDeletes);
      deletesObjectOutputStream.close();
      final int documentsBytesLength = documentsByteOutput.size();
      final int deletesBytesLength = deletesByteOutput.size();
      ByteBuffers documentsByteBuffers = documentsByteOutput.getByteBuffers();
      ByteBuffers deletesByteBuffers = deletesByteOutput.getByteBuffers();
      
      writeLock.lock();
      try {
        final long schemaVersion = masterBatch.getSchemaVersion();
        // TODO: add commit to logFileManager which adds record to the map if
        // successful transaction
        // TODO: multithread writerecord with http multicast
        // RecordHeader recordHeader = logFileManager.writeRecord(id, category,
        // schemaVersion, data);
        Future<RecordHeader> writeRecordFuture = threadPool.submit(new Callable<RecordHeader>() {
          public RecordHeader call() throws Exception {
            return logFileManager.writeRecord(id, category, schemaVersion, data);
          }
        });
        List<URL> urls = new ArrayList<URL>();
        HttpParameters httpParameters = new HttpParameters();
        httpParameters.add("action", "writetransaction");
        httpParameters.add("id", id);
        httpParameters.add("category", category);
        httpParameters.add("schemaversion", schemaVersion);
        httpParameters.add("previousid", previousId);
        List<NodeInfo> nodes = leases.getNodeInfos();
        for (NodeInfo node : nodes) {
          urls.add(new URL(node.getUrl().toString() + "?" + httpParameters.toEncodedQueryString()));
        }
        MulticastHttp multicastHttp = new MulticastHttp(threadPool, ExceptionEnum.ALL);
        HttpTimeout httpTimeout = new HttpTimeout(1000, 3 * 1000);
        try {
          List<OceanXML> responses = multicastHttp.executePost(urls, httpTimeout, new RequestEntity() {
            public void writeRequest(OutputStream output) throws IOException {
              DataOutputStream dataOutput = new DataOutputStream(output);
              dataOutput.writeInt(documentsBytesLength);
              documentsByteOutput.writeTo(dataOutput);
              dataOutput.writeInt(deletesBytesLength);
              deletesByteOutput.writeTo(dataOutput);
            }

            public long getContentLength() {
              return documentsBytesLength + deletesBytesLength;
            }

            public String getContentType() {
              return "binary/octet";
            }

            public boolean isRepeatable() {
              return true;
            }
          });
          for (OceanXML response : responses) {
            if (response.getHeaderBoolean("cancellease")) {
              String serverid = response.getHeaderValue("serverid");
              leases.cancelLease(serverid);
            } else if (response.getHeaderBoolean("renewlease")) {
              NodeInfo node = new NodeInfo(response.getFirstInBody());
              long duration = response.getHeaderLong("leaseduration");
              leases.renewLease(node, duration);
            }
          }
        } catch (AllExceptions allExceptions) {
          List<HttpRequestException> exceptions = allExceptions.getExceptions();
          if (exceptions.size() == nodes.size()) {
            // nothing worked
          }
          for (HttpRequestException requestException : exceptions) {
            URL url = requestException.getUrl();
            leases.cancelLease(url);
          }
          // TODO: throw exception
        } catch (Exception exception) {
          log.log(Level.SEVERE, "", exception);
          throw exception;
        }
        RecordHeader recordHeader = writeRecordFuture.get();
        logFileManager.commit(recordHeader);
        // TODO: commit recordheader
        // TODO: what if local write failed?
      } finally {
        writeLock.unlock();
      }
    } finally {
      documentsByteOutput.close();
      deletesByteOutput.close();
    }
  }
}
